package com.education.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.education.realtime.app.BaseAppV1;
import com.education.realtime.bean.TradeProvinceOrderWindow;
import com.education.realtime.common.Constant;
import com.education.realtime.fuction.DimAsyncFunction;
import com.education.realtime.util.FlinkSinkUtil;
import com.education.realtime.util.TimeUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Collections;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;

public class DwsTradeProvinceOrderWindow extends BaseAppV1 {
    public static void main(String[] args) {
        new DwsTradeProvinceOrderWindow().init(3001,"DwsTradeProvinceOrderWindow", Constant.TOPIC_DWD_TRADE_ORDER_DETAIL_APP);
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {

        //过滤去重数据
        SingleOutputStreamOperator<JSONObject> distinctStream = distinctOrderDetail(stream);
        //distinctStream.print();
        //封装成Javabean
        SingleOutputStreamOperator<TradeProvinceOrderWindow> parseToBeanStream = parseToBean(distinctStream);

        //按省份分组开窗聚合
        SingleOutputStreamOperator<TradeProvinceOrderWindow> windowAggrgate = windowAggrgate(parseToBeanStream);
       // windowAggrgate.print();

        //补充维度信息，省份名
        SingleOutputStreamOperator<TradeProvinceOrderWindow> joinDimStream = joinDim(windowAggrgate);
        joinDimStream.print();

        //写入clickhouse
        joinDimStream.addSink(FlinkSinkUtil.getClickHouseSink(Constant.CLICKHOUSE_DB,"dws_trade_province_order_window",TradeProvinceOrderWindow.class));

    }

    private SingleOutputStreamOperator<TradeProvinceOrderWindow> joinDim(SingleOutputStreamOperator<TradeProvinceOrderWindow> stream) {
        return AsyncDataStream.unorderedWait(
                stream,
                new DimAsyncFunction<TradeProvinceOrderWindow>() {
                    @Override
                    public void addDim(TradeProvinceOrderWindow input, JSONObject dim) {
                        input.setProvinceName(dim.getString("NAME"));
                    }

                    @Override
                    public String getTableName() {
                        return "dim_base_province";
                    }

                    @Override
                    public String getDimId(TradeProvinceOrderWindow input) {
                        return input.getProvinceId();
                    }
                },
                120,
                TimeUnit.SECONDS

        );

    }

    private SingleOutputStreamOperator<TradeProvinceOrderWindow> windowAggrgate(SingleOutputStreamOperator<TradeProvinceOrderWindow> stream) {
       return stream
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<TradeProvinceOrderWindow>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((obj,ts) -> obj.getTs()*1000))
                .keyBy(TradeProvinceOrderWindow::getProvinceId)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<TradeProvinceOrderWindow>() {
                            @Override
                            public TradeProvinceOrderWindow reduce(TradeProvinceOrderWindow t1, TradeProvinceOrderWindow t2) throws Exception {
                                t1.setOrderAmount(t1.getOrderAmount()+t2.getOrderAmount());
                                t1.getUserIdSet().addAll(t2.getUserIdSet());
                                t1.getOrderIdSet().addAll(t2.getOrderIdSet());
                                return t1;
                            }
                        },
                        new ProcessWindowFunction<TradeProvinceOrderWindow, TradeProvinceOrderWindow, String, TimeWindow>() {
                            @Override
                            public void process(String key,
                                                Context ctx,
                                                Iterable<TradeProvinceOrderWindow> iterable,
                                                Collector<TradeProvinceOrderWindow> out) throws Exception {
                                TradeProvinceOrderWindow bean = iterable.iterator().next();
                                bean.setStt(TimeUtil.toDateTime(ctx.window().getStart()));
                                bean.setEdt(TimeUtil.toDateTime(ctx.window().getEnd()));
                                bean.setUserCount((long) bean.getUserIdSet().size());
                                bean.setOrderCount((long) bean.getOrderIdSet().size());
                                bean.setTs(System.currentTimeMillis());

                                out.collect(bean);


                            }
                        }
                );



    }

    private SingleOutputStreamOperator<TradeProvinceOrderWindow> parseToBean(SingleOutputStreamOperator<JSONObject> stream) {
       return stream
                .map(obj ->
                        TradeProvinceOrderWindow.builder()
                                .provinceId(obj.getString("province_id"))
                                .userIdSet(new HashSet<>(Collections.singleton(obj.getString("user_id"))))
                                .orderIdSet(new HashSet<>(Collections.singleton(obj.getString("order_id"))))
                                .orderAmount(obj.getDoubleValue("final_amount"))
                                .ts(obj.getLong("ts"))
                                .build());
    }

    private SingleOutputStreamOperator<JSONObject> distinctOrderDetail(DataStreamSource<String> stream) {

       return stream
                .map(JSON::parseObject)
                .keyBy(obj -> obj.getString("id"))
                .process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                    private ValueState<JSONObject> lastOrderDetail;
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        lastOrderDetail = getRuntimeContext().getState(new ValueStateDescriptor<JSONObject>("lastOrderDetail", JSONObject.class));
                    }

                    @Override
                    public void processElement(JSONObject obj,
                                               Context ctx,
                                               Collector<JSONObject> out) throws Exception {
                        JSONObject value = lastOrderDetail.value();

                        if (value == null){
                            lastOrderDetail.update(obj);
                            out.collect(obj);
                        }

                    }
                });
    }
}
